package com.thunderbit.kafka.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * Kafka 配置类
 * 统一管理 Producer 和 Consumer 的配置
 */
public class KafkaConfig {

    // Kafka 服务器地址
    public static final String BOOTSTRAP_SERVERS = "localhost:9092";

    // 主题名称
    public static final String TOPIC_NAME = "demo-topic";

    // 消费者组 ID
    public static final String CONSUMER_GROUP_ID = "demo-consumer-group";

    /**
     * 获取 Producer 配置
     */
    public static Properties getProducerProperties() {
        Properties props = new Properties();

        // Kafka 服务器地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);

        // Key 序列化器
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());

        // Value 序列化器
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());

        // 确认机制：all 表示所有副本都确认
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        // 重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, 3);

        // 批量发送大小（字节）
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

        // 延迟时间（毫秒）
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);

        // 缓冲区大小（字节）
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

        return props;
    }

    /**
     * 获取 Consumer 配置
     */
    public static Properties getConsumerProperties() {
        Properties props = new Properties();

        // Kafka 服务器地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);

        // 消费者组 ID
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_ID);

        // Key 反序列化器
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());

        // Value 反序列化器
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());

        // 自动提交偏移量
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");

        // 自动提交间隔（毫秒）
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

        // 从最早的消息开始消费（如果没有已提交的偏移量）
        // 可选值：earliest（最早）、latest（最新）
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 一次拉取的最大记录数
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);

        return props;
    }
}